package com.gin.cep;

import com.gin.cep.vo.OrderInfo;
import com.gin.cep.vo.OrderMessage;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.util.List;

/**
 * 在京东里面，一个订单创建之后，15分钟内如果没有支付，会发送一个提示信息给用户，
 * 如果15分钟内已经支付的，需要发一个提示信息给商家
 *
 * @author gin
 * @date 2021/4/27
 */
public class OrderCepTest {

    public static void main(String[] args) {
        //获取运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置为事件时间时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        OutputTag<OrderMessage> payTimeoutTag = new OutputTag<OrderMessage>("pay_timeout") {
        };
        DataStreamSource<String> stream = env.readTextFile(Class.class.getResource("/OrderLog.csv").getPath());
        SingleOutputStreamOperator<OrderInfo> streamTime = stream
                .map(new RichMapFunction<String, OrderInfo>() {
                    @Override
                    public OrderInfo map(String resource) throws Exception {
                        String[] split = resource.split(",");
                        return new OrderInfo(split[0], split[1], split[2], Long.parseLong(split[3]));
                    }
                })
                .assignTimestampsAndWatermarks(//指定水印延迟10秒
                        new BoundedOutOfOrdernessTimestampExtractor<OrderInfo>(Time.seconds(10)) {
                            //将数据中的时间字段提取出来，转成Long类型,不改变输入的数据样式
                            @Override
                            public long extractTimestamp(OrderInfo loginEvent) {
                                return loginEvent.getActionTime() * 1000L;
                            }
                        });

        KeyedStream<OrderInfo, String> keyedStream = streamTime.keyBy(new KeySelector<OrderInfo, String>() {
            @Override
            public String getKey(OrderInfo orderInfo) throws Exception {
                return orderInfo.getOid();
            }
        });

        Pattern<OrderInfo, OrderInfo> pattern = Pattern.<OrderInfo>begin("myStart")
                .where(new IterativeCondition<OrderInfo>() {
                    @Override
                    public boolean filter(OrderInfo loginEvent, Context<OrderInfo> context) throws Exception {
                        return "create".equals(loginEvent.status);
                    }
                })
                //宽松邻近：在宽松邻近条件下，会忽略没有成功匹配模式条件
                //可以简单理解为OR的逻辑关系
                .followedBy("second")
                .where(new IterativeCondition<OrderInfo>() {
                    @Override
                    public boolean filter(OrderInfo loginEvent, Context<OrderInfo> context) throws Exception {
                        return "pay".equals(loginEvent.status);
                    }
                })
                //10秒内
                .within(Time.minutes(15));

        PatternStream<OrderInfo> ps = CEP.pattern(keyedStream, pattern);

        SingleOutputStreamOperator<OrderMessage> select = ps.select(payTimeoutTag,
                (PatternTimeoutFunction<OrderInfo, OrderMessage>) (map, l) -> {
                    List<OrderInfo> myStart = map.get("myStart");
                    OrderInfo orderInfo = myStart.get(0);
                    return new OrderMessage(orderInfo.oid, "该订单在15分钟内没有支付，请尽快支付", orderInfo.actionTime, 0);
                },
                (PatternSelectFunction<OrderInfo, OrderMessage>) map -> {
                    List<OrderInfo> myStart = map.get("myStart");
                    OrderInfo create = myStart.get(0);
                    List<OrderInfo> second = map.get("second");
                    OrderInfo pay = second.get(0);
                    return new OrderMessage(create.oid, "订单正常支付，请经过发货", create.actionTime, pay.actionTime);
                });

        select.getSideOutput(payTimeoutTag).print("侧流");
        select.print("主流");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
